<?php
namespace app\command;

use PhpMqtt\Client\ConnectionSettings;
use PhpMqtt\Client\MqttClient;
use think\console\Command;
use think\console\Input;
use think\console\input\Argument;
use think\console\input\Option;
use think\console\Output;
use app\admin\controller\Server;
use think\facade\Log;

cli_set_process_title('mqtt');
class Hello extends Command
{
    protected function configure()
    {
        $this->setName('hello')
            ->addArgument('name', Argument::OPTIONAL, "your name")
            ->addOption('city', null, Option::VALUE_REQUIRED, 'city name')
            ->setDescription('Say Hello');
    }
    protected function execute(Input $input, Output $output)
    {
        //创建Server对象，监听 127.0.0.1:9501 端口。
        $server = new \Swoole\Server('0.0.0.0', 9501);
        $server->set([  
            'daemonize' => true, // 设置为守护进程  
        ]);
        /*$server->set(array(
            'reactor_num'   => 2,     // 线程数
            'worker_num'    => 4,     // 进程数
            'backlog'       => 128,   // 设置Listen队列长度
            'max_request'   => 50,    // 每个进程最大接受请求数
            'dispatch_mode' => 1,     // 数据包分发策略
        ));*/
        /**
         * 用户进程实现了广播功能，循环接收unixSocket的消息，并发给服务器的所有连接
         */
        $process = new \Swoole\Process(function ($process) use ($server) {
            $ip   = '113.44.136.17';
            $port     = 1883;
            $clientId = rand(5, 15);
            $username = 'test';
            $password = '123456';
            $clean_session = false;
            $mqtt_version = MqttClient::MQTT_3_1_1;

            $connectionSettings = (new ConnectionSettings)
                ->setUsername($username)
                ->setPassword($password)
                ->setKeepAliveInterval(60)
                ->setLastWillTopic('testtopic/test')
                ->setLastWillMessage('client disconnect')
                ->setLastWillQualityOfService(1);


            $mqtt = new MqttClient($ip, $port, $clientId, $mqtt_version);

            $mqtt->connect($connectionSettings, $clean_session);
            //printf("client connected\n");

            $mqtt->subscribe('testtopic/#', function ($topic, $message) {
                //printf("Received message on topic [%s]: %s\n", $topic, $message);
                Log::write("日期：".date('Y-m-d H:i',time()." | topic:{$topic} | message:{$message}"),'notice');
                $model = new Server();
                $model->insert($topic,$message);
            }, 0);

            /*for ($i = 0; $i< 10; $i++) {
                $payload = array(
                    'protocol' => 'tcp',
                    'date' => date('Y-m-d H:i:s'),
                    'url' => 'https://github.com/emqx/MQTT-Client-Examples'
                );
                $mqtt->publish(
                // topic
                    'emqx/test',
                    // payload
                    json_encode($payload),
                    // qos
                    0,
                    // retain
                    true
                );
                printf("msg $i send\n");
                sleep(1);
            }*/

            
            while (true) {
                try {
                    $mqtt->loop(true);
                } catch (\Throwable $e) {
                    $mqtt->connect($connectionSettings, $clean_session);
                }
                //$mqtt->loop(true);
            }
        }, false, 2, 1);

        $server->addProcess($process);
        //监听连接进入事件。
        /**
         * reactor_id 线程id
         */
        $server->on('Connect', function ($server, $fd) {
            echo "Client: {$fd}Connect.\n";
        });

        //监听数据接收事件。
        $server->on('Receive', function ($server, $fd, $reactor_id, $data) {
            echo "收到的fd：".$fd."\n";
            $server->send($fd, "Server: {$reactor_id}-{$data}");
        });

        //监听连接关闭事件。
        $server->on('Close', function ($server, $fd) {
            echo "Client: Close.\n";
        });

        //启动服务器
        $server->start();



    }
}